package org.spbsu.apco.client;

import org.apache.cxf.binding.BindingFactoryManager;
import org.apache.cxf.interceptor.LoggingInInterceptor;
import org.apache.cxf.interceptor.LoggingOutInterceptor;
import org.apache.cxf.jaxrs.JAXRSBindingFactory;
import org.apache.cxf.jaxrs.client.JAXRSClientFactoryBean;
import org.apache.cxf.jaxrs.client.WebClient;
import org.perf4j.StopWatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.spbsu.apco.common.AuthorizationService;
import org.spbsu.apco.common.TaskService;
import org.spbsu.apco.common.monitoring.event.TaskEvent;
import org.spbsu.apco.common.properties.ApplicationPropertiesManager;
import org.spbsu.apco.common.properties.PropertyKey;
import org.spbsu.apco.common.serialization.Serializer;
import org.spbsu.apco.common.task.core.BaseTask;
import org.spbsu.apco.common.task.core.BatchTask;
import org.spbsu.apco.common.task.core.Ticket;
import org.spbsu.apco.common.task.core.WaitTask;
import org.spbsu.apco.common.utils.CalendarUtils;
import org.spbsu.apco.common.utils.NetworkUtils;
import org.spbsu.apco.common.utils.StopWatchFactory;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import javax.ws.rs.core.MultivaluedMap;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * User: solozobov
 * Date: 11.11.12
 */

public class Client implements Runnable {

    private static final Logger LOG = LoggerFactory.getLogger(Client.class);
    private static final int INFINITE_LOOP_SEEP_INTERVAL = 10 * 60 * 1000; // 10 min
    private Integer DISPATCHING_TIMEOUT;

    private final ThreadPoolExecutor executor;
    private TaskService taskService;
    private String clientId = null;
    private Logger log;

    private Serializer serializer;

    public Client() {
        final Integer threadsCount = Integer.valueOf(ApplicationPropertiesManager.getProperty(PropertyKey.WORKING_THREADS_COUNT));
        executor = new ClientThreadPoolExecutor(threadsCount);
        final String brokerUrl = ApplicationPropertiesManager.getProperty(PropertyKey.BROKER_URL);

        final AuthorizationService authorizationService = createTaskService(AuthorizationService.class, brokerUrl);
        retryUntilAuthorized(threadsCount, brokerUrl, authorizationService);

        taskService = createTaskService(TaskService.class, brokerUrl);
        log = LoggerFactory.getLogger(Client.class.getSimpleName() + clientId);
    }

    private void retryUntilAuthorized(Integer threadsCount, String brokerUrl, AuthorizationService authorizationService) {
        do {
            try {
                clientId = authorizationService.getId(threadsCount, null);
            } catch (Exception e) {
                LOG.info("Authentication failed. Retry to connect: " + brokerUrl);
                if (DISPATCHING_TIMEOUT == null) {
                    DISPATCHING_TIMEOUT = Integer.parseInt(ApplicationPropertiesManager.getProperty(PropertyKey.DISPATCHING_TIMEOUT)) * 1000; // convert from sec to ms
                }
                try {
                    Thread.sleep(DISPATCHING_TIMEOUT);
                } catch (InterruptedException ignored) {
                    //should not happen
                }
            }
        } while (clientId==null);
    }

    public void init() {
        final Thread clientThread = new Thread(this, "Client-service-" + this.clientId);
        clientThread.setDaemon(true);
        clientThread.start();
    }

    @SuppressWarnings("InfiniteLoopStatement")
    @Override
    public void run() {
        while (true) {
            boolean needToWait = true;

            final StopWatch stopWatch = StopWatchFactory.createOrGetLocal();
            stopWatch.start();
            final byte[] response = taskService.getTask(clientId);
            stopWatch.lap("client__task-transfer");
            final Ticket ticket = serializer.deserializeTaskAndInput(response);
            stopWatch.lap("client__task-deserialization");
            final BaseTask<?, ?> task = ticket.getTask();
            if (task instanceof WaitTask) {
                stopWatch.stop("client__task-scheduling");
                execute(this, task, ticket.getArguments());
                needToWait = false;
            } else if (task instanceof BatchTask) {
                log.info("Client #" + clientId + " accepted BatchTask, schedules its arguments now...");
                for (Ticket subTicket : (List<Ticket>) ticket.getArguments()) {
                    schedule(subTicket.getTask(), subTicket.getArguments());
                }
                stopWatch.stop("client__task-scheduling");
            } else {
                final List<?> arguments = ticket.getArguments();
                schedule(task, arguments);
                stopWatch.stop("client__task-scheduling");
            }

            try {
                if (needToWait) {
                    synchronized (executor) {
                        executor.wait();
                    }
                }
            } catch (InterruptedException e) {
                LOG.error("Interrupted", e);
            }
        }
    }

    private void schedule(final BaseTask<?, ?> task, final List<?> arguments) {
        log.info("Client #" + clientId + " scheduled for execution task #" + task.getId());
        executor.execute(new Runnable() {
            @Override
            public void run() {
                execute(Client.this, task, arguments);
            }
        });
    }


    private void setTaskEventHeader(BaseTask<?, ?> task) {
        org.apache.cxf.jaxrs.client.Client client = WebClient.client(taskService);
        TaskEvent taskEvent = new TaskEvent();
        taskEvent.setOriginatorId(this.clientId);
        taskEvent.setTaskStartTime(task.getTaskStartTime());
        taskEvent.setTaskDuration(task.getTaskDuration());
        taskEvent.setTaskId("" + task.getId());
        taskEvent.setTaskType(task.getType());
        List<TaskEvent> taskList = Arrays.asList(taskEvent);
        final String encodedEvent = new String(serializer.encodeTaskEvents(taskList));
        client.header(NetworkUtils.X_EVENT, encodedEvent);
    }

    private void removeTaskEventHeader() {
        org.apache.cxf.jaxrs.client.Client client = WebClient.client(taskService);
        MultivaluedMap<String, String> allHeaders = client.getHeaders();
        client.reset();
        allHeaders.remove(NetworkUtils.X_EVENT);
        client.headers(allHeaders);
    }

    private void execute(Client client, BaseTask<?, ?> task, List<?> arguments) {
        client.log.info("Client #" + client.clientId + " started task #" + task.getId() + " " + task.getClass().getSimpleName());
        final StopWatch stopWatch = StopWatchFactory.createOrGetLocal();
        stopWatch.start();
        final Object result = task.runTask(arguments);
        stopWatch.lap("client__task-execution");
        client.setTaskEventHeader(task);
        stopWatch.lap("client__task-headers");
        final byte[] resultRepresentation = serializer.serializeResult(result);
        stopWatch.lap("client__task-serialization");
        client.taskService.holdResult(client.clientId, task.getId(), resultRepresentation);
        stopWatch.lap("client__task-result-transferring");
        client.removeTaskEventHeader();
        stopWatch.lap("client__task-remove-headers");
        client.log.info("Client #" + client.clientId + " processed task #" + task.getId() + " and sent result");
    }

    public static void main(String[] args) {
        start();
        //noinspection InfiniteLoopStatement
        while (true) {
            //noinspection EmptyCatchBlock
            try {
                Thread.sleep(INFINITE_LOOP_SEEP_INTERVAL);
            } catch (InterruptedException e) {
            }
        }
    }


    public static void start() {
        new ClassPathXmlApplicationContext("client.xml");
    }


    private static <T> T createTaskService(Class<T> serviceClass, String brokerUrl) {
        final JAXRSClientFactoryBean sf = new JAXRSClientFactoryBean();
        sf.setResourceClass(serviceClass);
        sf.setAddress(brokerUrl);

        if (LOG.isDebugEnabled()) {
            sf.getInInterceptors().add(new LoggingInInterceptor());
            sf.getOutInterceptors().add(new LoggingOutInterceptor());
        }

        final BindingFactoryManager manager = sf.getBus().getExtension(BindingFactoryManager.class);

        final JAXRSBindingFactory factory = new JAXRSBindingFactory();
        factory.setBus(sf.getBus());

        manager.registerBindingFactory(JAXRSBindingFactory.JAXRS_BINDING_ID, factory);

        final T proxyService = sf.create(serviceClass);

        final org.apache.cxf.jaxrs.client.Client client = WebClient.client(proxyService);
        try {
            client.header(NetworkUtils.X_MAC_SRC, NetworkUtils.getHostname() + "@" + CalendarUtils.getTimeZoneId());
        } catch (Exception e) {
            throw new IllegalStateException("Cannot define local MAC address", e);
        }
        return proxyService;
    }

    public void setSerializer(final Serializer serializer) {
        this.serializer = serializer;
    }

}
